package top.scauwlt.remoting.transport.server;

import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandler;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;
import top.scauwlt.config.RpcConfig;
import top.scauwlt.enums.CompressTypeEnum;
import top.scauwlt.enums.SerializationTypeEnum;
import top.scauwlt.factory.SingletonFactory;
import top.scauwlt.remoting.constants.RpcConstants;
import top.scauwlt.remoting.dto.RpcMessage;
import top.scauwlt.remoting.dto.RpcRequest;
import top.scauwlt.remoting.dto.RpcResponse;

/**
 * @Author wmm
 * @Date 2022/7/26 17:34
 */
@Slf4j
public class NettyRpcServerHandler extends ChannelInboundHandlerAdapter {

    private final RpcRequestHandler rpcRequestHandler;

    public NettyRpcServerHandler() {
        rpcRequestHandler = SingletonFactory.getInstance(RpcRequestHandler.class);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        try {
            if (msg instanceof RpcMessage) {
                log.info("server receive msg: [{}]", msg);
                //创建回复消息
                RpcMessage response = new RpcMessage();
                response.setCodec(SerializationTypeEnum.getCode(RpcConfig.getInstance().getSerializationType()));
                response.setCompress(CompressTypeEnum.getCode(RpcConfig.getInstance().getCompressType()));
                byte messageType = ((RpcMessage) msg).getMessageType();
                if (messageType == RpcConstants.HEARTBEAT_REQUEST_TYPE) {
                    response.setMessageType(RpcConstants.HEARTBEAT_RESPONSE_TYPE);
                    response.setData(RpcConstants.PONG);
                } else {
                    response.setMessageType(RpcConstants.RESPONSE_TYPE);
                    RpcRequest rpcRequest = (RpcRequest) ((RpcMessage) msg).getData();
                    Object result = rpcRequestHandler.handle(rpcRequest);
                    log.info("server get result: {}", result);
                    response.setMessageType(RpcConstants.RESPONSE_TYPE);
                    if (ctx.channel().isActive() && ctx.channel().isWritable()) {
                        RpcResponse<Object> data = RpcResponse.success(result, rpcRequest.getRequestId());
                        response.setData(data);
                    } else {
                        RpcResponse<Object> data = RpcResponse.fail(result, rpcRequest.getRequestId());
                        response.setData(data);
                        log.error("server not writable now, message dropped: [{}]", response);
                    }
                }
                ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            }
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.READER_IDLE) {
                log.info("over 30s without read event, close the connection");
                ctx.close();
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("server catch exception");
        cause.printStackTrace();
        ctx.close();
    }
}
